Add broker startup pre-connect for broker-to-server channels (SSE) - #19407
Add broker startup pre-connect for broker-to-server channels (SSE)#19407jineshparakh wants to merge 1 commit into
Conversation
Signed-off-by: Jinesh Parakh <jineshparakh@hotmail.com>
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19407 +/- ##
============================================
- Coverage 67.56% 67.55% -0.01%
Complexity 1430 1430
============================================
Files 3486 3487 +1
Lines 224173 224273 +100
Branches 35381 35397 +16
============================================
+ Hits 151462 151512 +50
- Misses 60684 60721 +37
- Partials 12027 12040 +13
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
gortiz
left a comment
gortiz
left a comment
There was a problem hiding this comment.
Thanks for this — the problem is real and well diagnosed, and I appreciate that the description comes with measurements from a TLS-enabled cluster rather than a microbenchmark. The ServerPreConnector seam (dependencies as functions) is a nice touch that makes the budget and failure handling genuinely unit-testable. I'd like to see it land, but there are a few things I think need addressing first.
Leaving this as a comment rather than a formal request for changes — none of it is a disagreement with the goal.
The items I'd want resolved are inline. Briefly:
- The lazy query path silently loses the
NETTY_CONNECTION_CONNECT_TIME_MSgauge, which contradicts the "behaviour identical to before" claim forenabled=false. ServerChannels.connect()is also the failure detector's reconnect probe, not only startup pre-connect (the javadoc says otherwise). It now holds the channel lock across the TLS handshake on a path that runs under live traffic.hasChannel()becomes unconditionally true once pre-connect has run, which kills theUNKNOWNescape hatch that keeps MSE-only clusters out of the SSE health state machine.MAX_CONNECT_THREADS = 16combined with Netty's default 30s connect timeout means a handful of black-holed servers can consume the entire pre-connect budget and connect nothing.- Connecting both table types unconditionally doubles channel count on single-type clusters.
_channelis assigned before the handshake is awaited, so a failed handshake leaves the field pointing at a doomed-but-briefly-active channel.
Two larger questions I'd like your view on, which I haven't left inline because they're about the shape of the change rather than a specific line:
-
Is the readiness gate worth it? The measured win is ~200ms on the first query per server. The price is that the broker's only health endpoint (
/health— there's no liveness/readiness split on the broker) returns 503 for up to 30s, on a duration that depends on remote server reachability. The bundled Helm chart does define astartupProbe, so it's safe there, but the "no probe change is required" claim only holds for deployments that have one; anyone wiringlivenessProbe->/healthwith the commonperiodSeconds=10, failureThreshold=3gets a 30s window that can turn a warm-up into a restart loop. My inclination would be to keep the warm-up and drop the gate, or default the flag tofalse. -
Is one-shot warm-up the right scope? Pre-connect fires once, with no retry. On a full-cluster cold start (brokers and servers booting together) every connect gets
ECONNREFUSEDin milliseconds, soconnected == 0and the gate is paid for nothing — the feature really only helps a broker-only rolling restart against already-live servers. It also doesn't cover servers that join later (scale-up, server restart). If the goal is "connect is never on the query path", hooking routing changes to warm new servers — or moving channel establishment out of thesendRequestlock into a shared per-channel future — would cover all three cases instead of one.
On tests: in BrokerServerPreConnectIntegrationTest, startBroker() runs before startServer(), so at Helix convergence getRoutableServerInstanceMap() is empty and preConnect() returns 0 immediately. preConnectEnabledBrokerReachesGoodServiceStatus therefore asserts GOOD against a gate that was never actually held — it would pass with the feature stubbed out. The readiness gate is the highest-risk part of this change and nothing currently covers it; starting the server first, or injecting a slow connect function, would make the STARTING window observable and assertable.
One naming nit while things are still movable: BrokerTimer.NETTY_CONNECTION_CONNECT_TIME omits the _MS suffix that both of its neighbours carry, and sits next to a gauge whose name differs only by that suffix. Metric names are effectively permanent, so worth fixing before merge.
| @@ -236,11 +237,45 @@ void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse, | |||
|
|
|||
| void connectWithoutLocking() | |||
There was a problem hiding this comment.
The lazy query path no longer records BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS. Before this PR, every connect set that gauge; now only connectAndAwaitHandshakeWithoutLocking() does, and sendRequest() -> connectWithoutLocking() records nothing.
That makes the backward-compatibility claim inaccurate: with preconnect.enabled=false — documented as "a strict no-op, behaviour identical to before" — this is the only connect path, so any existing dashboard or alert on NETTY_CONNECTION_CONNECT_TIME_MS goes permanently flat. And even with the flag on, the gauge's meaning silently narrows from "time to establish any channel" to "time to establish a pre-connect / failure-detector channel".
Suggest keeping the gauge (and ideally the new timer) in connectWithoutLocking() as well. Two System.currentTimeMillis() calls are not what makes the critical section long — the sync() is.
| throws InterruptedException { | ||
| SslHandler sslHandler = channel.pipeline().get(SslHandler.class); | ||
| if (sslHandler != null) { | ||
| sslHandler.handshakeFuture().sync(); |
There was a problem hiding this comment.
handshakeFuture().sync() rethrows a handshake failure after _channel has already been assigned on line 256. The field is then left pointing at a channel whose handshake failed, and until Netty finishes closing it, _channel.isActive() can still return true — at which point connectWithoutLocking() will write a query into it.
Narrow race, but cheap to close: assign _channel only after awaitTlsHandshake succeeds, or close/null it in a catch before propagating.
Also worth reflecting in the javadoc: sync() here is bounded by Netty's default SslHandler handshake timeout (10s), not by the caller's deadline, so one hung TLS peer parks a pre-connect worker for 10s regardless of the configured budget.
| if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { | ||
| try { | ||
| connectWithoutLocking(); | ||
| connectAndAwaitHandshakeWithoutLocking(); |
There was a problem hiding this comment.
The javadoc on connectAndAwaitHandshakeWithoutLocking() says it is "used only by startup pre-connect", but ServerChannels.connect() is also the failure detector's reconnect probe: SingleConnectionBrokerRequestHandler.retryUnhealthyServer() -> QueryRouter.connect() -> here. That path runs at steady state, under live traffic.
So this change means the probe now holds _channelLock across the entire TLS handshake. sendRequest() acquires the same lock with tryLock(queryTimeoutMs), so concurrent queries to a server that just came back queue behind the handshake — bounded only by Netty's default 10s handshakeTimeoutMillis, since ChannelHandlerFactory.getClientTlsHandler doesn't set one. That is the same serialization this PR sets out to remove, reintroduced on the runtime path.
Suggest splitting the two callers: keep connect() on the TCP-only variant, and give startup pre-connect its own entry point that awaits the handshake.
| /// single channel rather than every channel the server may need. Callers that want the server fully | ||
| /// connected -- startup pre-connect, for instance -- should use [#connect(ServerInstance, TableType)] | ||
| /// for each table type instead. | ||
| public boolean connect(ServerInstance serverInstance) { |
There was a problem hiding this comment.
A side effect that isn't called out in the description: pre-connect calls ServerChannels.connect() for every routable server, and that does computeIfAbsent on _serverToChannelMap — the entry is created even when the connect itself fails. Since hasChannel() (line 149) tests for the OFFLINE entry, it becomes unconditionally true after pre-connect.
That kills the escape hatch at SingleConnectionBrokerRequestHandler.retryUnhealthyServer():436, which returns ServerState.UNKNOWN when !hasChannel(serverInstance) specifically so an MSE-only cluster doesn't drive its servers through the SSE-channel health state machine. With pre-connect on (the default) that branch is dead, and MSE-only clusters will start marking servers UNHEALTHY based on SSE reachability.
Either outcome may be defensible, but it should be a deliberate decision with a test pinning it.
| int connected = 0; | ||
| try { | ||
| for (ServerInstance server : servers) { | ||
| for (TableType tableType : TableType.values()) { |
There was a problem hiding this comment.
Connecting both TableType values unconditionally doubles the channel count regardless of what the cluster actually routes. On an offline-only cluster, every broker opens and then holds N idle REALTIME TLS connections that will never carry a query — and pays N extra handshakes, on both ends, at every broker restart. ServerChannel entries are never evicted from _serverToChannelMap, so they persist for the process lifetime.
RoutingManager already knows which table types route to which server. Deriving the (server, tableType) pairs from routing instead of taking the cross product would be exactly right on hybrid clusters and halve the work on single-type ones. It also shrinks the hasChannel() side effect noted on QueryRouter.
| } | ||
| long startMs = System.currentTimeMillis(); | ||
| int channelCount = servers.size() * TableType.values().length; | ||
| ExecutorService executor = Executors.newFixedThreadPool(Math.min(channelCount, MAX_CONNECT_THREADS), |
There was a problem hiding this comment.
MAX_CONNECT_THREADS = 16 acts as a starvation cliff here, not just a throughput cap. The Bootstrap sets no ChannelOption.CONNECT_TIMEOUT_MILLIS (ServerChannels.java:179-180), so Netty's 30s default applies — which is exactly DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS. Sixteen servers whose SYN is dropped rather than refused (booting, or a security-group/firewall black hole) park the entire pool for the whole budget: zero channels connected, and the readiness gate held the full 30s having achieved nothing.
ECONNREFUSED returns in microseconds, so ordinary cold start is fine — but the case the 30s budget exists for is precisely the one where the thread count is the binding constraint. Setting CONNECT_TIMEOUT_MILLIS on the pre-connect path, so a single connect can't outlive the budget, would bound this independently of the pool size.
Relatedly, the comment on line 95 says "no head-of-line blocking". That's true of the ExecutorCompletionService, which removes head-of-line blocking from the counting; it doesn't remove it from execution, where the fixed workers are the queue. Worth rewording so it doesn't read as a stronger guarantee than it makes.
3 participants
Summary
On a freshly (re)started broker, the broker to server Netty channels start empty. The first query to
each server therefore pays the blocking TCP
connect(), and, when broker to server TLS is enabled, thefull TLS handshake, on its own critical path while holding the per server channel lock. Under a cold
burst of concurrent queries this serializes: every request thread targeting that server blocks on the one
in flight connect, and the TLS handshake (two round trips plus certificate validation) lengthens the
critical section.
This PR opens all broker to server channels ahead of query traffic during startup, behind a readiness
gate. When it is enabled the broker reports
STARTINGuntil the channels are connected (or a budgetexpires), so no traffic is routed to it until the connect and handshake cost has already been paid off the
query path. It is a startup only, best effort warmup: on by default, bounded so it can never stall a
rolling restart, and a no op reversion to the existing lazy connect path when disabled.
What it does
exist), a background thread opens a channel to every routable server, for both table types.
ServerRoutingInstanceidentity includes the table type, so OFFLINE and REALTIME are separate channelsto the same physical server.
(
SslHandler.handshakeFuture().sync()), so the handshake, not just the TCP connect, is off the firstquery's critical path. On a plaintext channel there is no
SslHandlerin the pipeline and this step isa no op.
ServiceStatusreportsSTARTING(an existingstatus value, so no new enum and mixed version peers are unaffected) until pre-connect finishes. This
reuses the existing readiness endpoint that the Kubernetes startup probe already polls; no health
endpoint or probe change is required.
opens even if pre-connect throws, is interrupted, or the budget expires, so a slow or unreachable
server can never hold a broker not ready forever. A channel that fails to connect simply falls back to
the existing lazy path. Connecting an already active channel is a no op, so this is idempotent.
This is single stage (SSE) only. The multi stage (gRPC) and time series paths use a different
transport and are unaffected.
Configuration
pinot.broker.startup.preconnect.enabledtruefalseto restore the pure lazy connect path (behaviour then unchanged).pinot.broker.startup.preconnect.timeoutMs30000Metrics
STARTUP_PRECONNECT_DURATION_MSNETTY_CONNECTION_CONNECT_TIMEPerformance
Setup
All numbers below are from a TLS-enabled single broker test cluster (a real ZK plus controller plus
server plus broker, not a synthetic mock):
nettytls), so the channels pre-connectopens pay a real TLS handshake. This is exactly the cost the fix targets.
replica. It routes to the one server, so the broker holds 2 channels to it (OFFLINE and REALTIME).
arrival, no ramp). Because arrivals do not wait for responses, under elevated cold latency the in flight
count builds to several hundred concurrent requests, which is what makes the per server connect and
handshake serialize.
pinot.broker.startup.preconnect.enabledOFF vs ON), n=10 cold restarts each:a per-query
serverStatscomparison of the first-query legs, and an open-loop run split into a coldwindow
[0, 8 s)and a warm window[150, 180 s)measured from the first traffic at readiness,with wall and lock profiles captured.
Broker pre-connected 2/2 channel(s) in 70 to 121 ms, then readiness.Gain
Pre-connect removes the connect and TLS handshake from the first query's critical path. Measured directly
from the first query's
serverStatslegs (median [range] across 10 cold restarts per arm):The broker to server TLS handshake for the 2 channels is now paid once at startup
(
pre-connected 2/2 channel(s) in 70 to 121 ms), off the query path, rather than by the first queryholding the channel lock. This is corroborated by the wall profile: the broker to server
connect/handshake park category (
QueryRouter.submitQuerytoServerChannels.sendRequestto park) is theonly category that changes between arms, dropping from 0.23% to 0.09% (about 2.5x). By readiness
_channel.isActive()istrue, so the lazyconnectWithoutLockingpath no ops and the first queryserialization never happens.
Readiness cost is negligible: the gate adds effectively no startup time (both arms reached readiness in
about 90 s), and it opens even on failure, so it cannot delay a rolling restart beyond the configured
budget.
Testing
ServerPreConnectorTest, 6 tests): connects every server for both table types; empty serverlist is a no op; already passed deadline is a no op; counts only successful connects; a throwing connect
is swallowed and the others still connect; the budget bounds the wait and does not block on slow
connects. The connector takes its dependencies (routable server supplier, connect function) as
functions, so parallelism, budget, and failure handling are covered without a live broker.
BrokerServerPreConnectIntegrationTest): brings up a real ZK plus controller plusserver plus broker with an offline table and asserts (1) with pre-connect enabled the readiness gate
opens, the broker's
ServiceStatusreachesGOOD; and (2) the production path (RoutingManagerroutable server supplier to
QueryRoutertoServerChannelsto a live server) opens one channel per(server, table type).
Backward compatibility
STARTINGstatus, with no new status enum value, so mixed versionbroker/controller peers are unaffected.
preconnect.enabled=falsethe code path is a strict no op and behaviour is identical to before.